<?php

namespace Diablo7\HyperfPulsar;

use Diablo7\HyperfPulsar\Message\ConsumerMessageInterface;
use Diablo7\HyperfPulsar\Message\MessageInterface;
use Hyperf\Contract\ContainerInterface;
use Hyperf\Pool\Pool;

class Pulsar
{
    const ACK = 1;
    const NACK = 2;
    /**
     * @var ContainerInterface
     */
    public $container;

    /**
     * @var Pool[]
     */
    public $pools = [];

    public function __construct(ContainerInterface $container)
    {
        $this->container = $container;
    }

    public function getPool(MessageInterface $message)
    {
        $key = get_class($message);
        if(!isset($this->pools[$key])){
            $this->pools[$key] =  new PulsarPool($this->container,$message);
        }
        return $this->pools[$key];
    }

    public function consume(ConsumerMessageInterface $message)
    {
        $connection = $this->getPool($message)->get();

        //TODO 消息消费的实现
        try {

            $connection->call(function (PulsarConnection $con) use($message){

                $con->client->recv();


            });

        }catch (\Exception $exception){

        }


    }

    public function produce(MessageInterface $message)
    {
        $connection = $this->getPool($message)->get();

        //TODO 消息生产的实现
        $result = $connection->call(function ($conn){

        });


    }


    public function read(MessageInterface $message)
    {

    }
}